package rabbit_mq

import (
	"context"
	"crypto/tls"
	"fmt"
	"time"

	"code.byted.org/gopkg/logs"

	"github.com/pkg/errors"

	amqp "github.com/rabbitmq/amqp091-go"
)

func NewClient(config Config) (*Client, error) {
	mq := &Client{
		config:      config,
		isConnected: false,
	}
	if err := mq.connect(); err != nil {
		return nil, err
	}

	go mq.handleReconnect()
	return mq, nil
}

type Client struct {
	config          Config
	queueName       string
	connection      *amqp.Connection
	channel         *amqp.Channel
	chanNotifyClose chan *amqp.Error
	connNotifyClose chan *amqp.Error
	notifyConfirm   chan amqp.Confirmation
	isConnected     bool
}

func (mq *Client) handleReconnect() {
	for {
		select {
		case <-mq.chanNotifyClose:
			logs.Warn("rabbit mq chan disconnect,attempt to reconnect ...")
			mq.isConnected = false
			for {
				if err := mq.connect(); err != nil {
					time.Sleep(time.Second)
				} else {
					break
				}
			}
		case <-mq.connNotifyClose:
			logs.Warn("rabbit mq chan disconnect,attempt to reconnect ...")
			mq.isConnected = false
			for {
				if err := mq.connect(); err != nil {
					time.Sleep(time.Second)
				} else {
					break
				}
			}
		}
	}
}

func (mq *Client) connect() error {
	c := &tls.Config{
		InsecureSkipVerify: true, //todo rabbitmq修复上线后去掉
	}
	conn, err := amqp.DialTLS(mq.config.genUrlWithSsl(), c)
	if err != nil {
		return errors.WithStack(err)
	}

	ch, err := conn.Channel()
	if err != nil {
		return errors.WithStack(err)
	}

	queue, err := ch.QueueDeclare(
		mq.config.Queue,
		true,
		false,
		false,
		false,
		amqp.Table{
			"x-max-priority": 10,
		})
	if err != nil {
		return errors.WithStack(err)
	}

	mq.changeConnection(conn, ch)
	mq.queueName = queue.Name
	mq.isConnected = true
	logs.Info("rabbit mq connect success")
	return nil
}

func (mq *Client) changeConnection(connection *amqp.Connection, channel *amqp.Channel) {
	mq.connection = connection
	mq.channel = channel
	mq.chanNotifyClose = make(chan *amqp.Error, 10)
	mq.connNotifyClose = make(chan *amqp.Error, 10)
	mq.notifyConfirm = make(chan amqp.Confirmation)
	mq.channel.NotifyClose(mq.chanNotifyClose)
	mq.connection.NotifyClose(mq.connNotifyClose)
}

func (mq *Client) Publish(ctx context.Context, msg []byte, priority Priority) error {
	if !mq.isConnected {
		return errors.New("temporarily unavailable,please try again")
	}
	err := mq.channel.PublishWithContext(ctx, "", mq.queueName, false, false, amqp.Publishing{
		Headers:         amqp.Table{},
		ContentType:     "application/json",
		ContentEncoding: "",
		Body:            msg,
		DeliveryMode:    amqp.Persistent, // 1=non-persistent, 2=persistent
		Priority:        uint8(priority), // 0-9
	})
	return errors.WithStack(err)
}

type Config struct {
	Host     string `json:"host" yaml:"Host" viper:"Host"`
	Port     string `json:"port" yaml:"Port" viper:"Port"`
	Username string `json:"username" yaml:"Username" viper:"Username"`
	Password string `json:"password" yaml:"Password" viper:"Password"`
	Vhost    string `json:"vhost" yaml:"Vhost" viper:"Vhost"`
	Queue    string `json:"queue" yaml:"Queue" viper:"Queue"`
}

func (c *Config) genUrlWithSsl() string {
	return fmt.Sprintf("amqps://%s:%s@%s:%s%s", c.Username, c.Password, c.Host, c.Port, c.Vhost)
}
